﻿//------------------------------------------------------------------------------
// <copyright file="RpcChannel.cs" company="Microsoft">
//     Copyright (c) Microsoft Corporation.  All rights reserved.
// </copyright>
//------------------------------------------------------------------------------

namespace Microsoft.Internal.GamesTest.Rpc.Client
{
    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Diagnostics;
    using System.Globalization;
    using System.IO;
    using System.Linq;
    using System.Net;
    using System.Reflection;
    using System.Runtime.Serialization;
    using System.Runtime.Serialization.Json;
    using System.Security.Principal;
    using System.Text;
    using System.Threading;
    using System.Threading.Tasks;
    using Microsoft.Internal.GamesTest.Rpc.Client.Telemetry;
    using Microsoft.Internal.GamesTest.Rpc.Client.Transport;

    /// <summary>
    /// This is the primary entry point into to the TDK RPC client-side library.
    /// This class is used to the RpcServer server and invoke the methods that it exposes.
    /// </summary>
    public class RpcChannel : DisposableObject
    {
        private const string PlatformValue = "Platform";
        private const string ArchitectureValue = "Architecture";

        private static readonly Guid DefaultServerId = Guid.Parse("21A25F93-2841-4CC7-A380-FD91ABBE43D1");
        private static readonly TimeSpan DefaultOperationTimeout = TimeSpan.FromSeconds(30.0);
        private static readonly Type[] JsonSerializerKnownTypes = new Type[] 
            { 
                typeof(sbyte[]),
                typeof(byte[]),
                typeof(short[]),
                typeof(ushort[]),
                typeof(int[]),
                typeof(uint[]),
                typeof(long[]),
                typeof(ulong[]),
                typeof(float[]),
                typeof(double[]),
                typeof(decimal[]),
                typeof(string[]),
                typeof(RpcArchive),
                typeof(RpcBinaryData)
            };

        private static long lastRequestId = 0;

        private DataContractJsonSerializer responseDeserializer;
        private DataContractJsonSerializer messageSerializer;

        private RpcSocketTransport transport;
        private ConcurrentDictionary<string, ConcurrentDictionary<Guid, EventHandler<RpcEventArgs>>> serverEventSubscriptions;

        private ConcurrentDictionary<long, AsyncResult<RpcResponse>> pendingRequests;

        private Task readingIncomingMessagesTask;
        private CancellationTokenSource cancelReadingTokenSource;

        private IExceptionFactory exceptionFactory;

        private Guid connectionId;

        private ReaderWriterLockSlim connectLock = new ReaderWriterLockSlim();

        /// <summary>
        /// Initializes static members of the RpcChannel class.
        /// </summary>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Performance", "CA1810:InitializeReferenceTypeStaticFieldsInline", Justification = "Since this static constructor is doing more than initializing static fields, but is actually effecting global state, this rule is appropriate to suppress.")]
        static RpcChannel()
        {
            // Wait for a short time for telemetry initialization to finish
            // so we can try to get telemetry for the first few Rpc activites.
            // then whether or not it's finished go ahead and return control
            // to users to keep Rpc responsive.
            Task.Run(() =>
                {
                    TelemetrySink.StartTelemetry();

                    // Initialization of telemetry finished.
                    AppDomain.CurrentDomain.ProcessExit += (sender, args) =>
                    {
                        TelemetrySink.StopTelemetry();
                    };

                    RpcClientEventSource.Logger.ModuleLoaded(
                        Process.GetCurrentProcess().ProcessName,
                        WindowsIdentity.GetCurrent().Name,
                        Dns.GetHostEntry("localhost").HostName,
                        FileVersionInfo.GetVersionInfo(Assembly.GetExecutingAssembly().Location).FileVersion.ToString());
                    
                    TelemetrySink.Flush();
                })
                .Wait(TimeSpan.FromMilliseconds(100));
        }

        /// <summary>
        /// Initializes a new instance of the RpcChannel class.
        /// </summary>
        /// <param name="ipAddress">The IP address of the server to connect to.</param>
        /// <param name="serverId">The Id of the RpcServer running on the server side.</param>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Reliability", "CA2000:Dispose objects before losing scope", Justification = "The object will be disposed in this class' dispose method.")]
        public RpcChannel(IPAddress ipAddress, Guid serverId)
        {
            RpcClientEventSource.Logger.ObjectCreated(RpcClientEventSource.GetCurrentConstructor(serverId == DefaultServerId));
            if (ipAddress == null)
            {
                throw new ArgumentNullException("ipAddress");
            }

            this.Initialize(new RpcSocketTransport(ipAddress, serverId));
        }

        /// <summary>
        /// Initializes a new instance of the RpcChannel class with port number.
        /// </summary>
        /// <param name="ipAddress">The IP address of the server to connect to.</param>
        /// <param name="port">The Port number on the server to connect to.</param>
        /// <param name="serverId">The Id of the RpcServer running on the server side.</param>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Reliability", "CA2000:Dispose objects before losing scope", Justification = "The object will be disposed in this class' dispose method.")]
        public RpcChannel(IPAddress ipAddress, int port, Guid serverId)
        {
            RpcClientEventSource.Logger.ObjectCreated(RpcClientEventSource.GetCurrentConstructor(serverId == DefaultServerId));
            if (ipAddress == null)
            {
                throw new ArgumentNullException("ipAddress");
            }

            this.Initialize(new RpcSocketTransport(ipAddress, port, serverId));
        }

        /// <summary>
        /// Initializes a new instance of the RpcChannel class.
        /// </summary>
        /// <param name="ipAddress">The IP address of the server to connect to.</param>
        public RpcChannel(IPAddress ipAddress)
            : this(ipAddress, DefaultServerId)
        {
        }

        /// <summary>
        /// Initializes a new instance of the RpcChannel class with port number.
        /// </summary>
        /// <param name="ipAddress">The IP address of the server to connect to.</param>
        /// <param name="port">The port number on the server to connect to.</param>
        public RpcChannel(IPAddress ipAddress, int port)
            : this(ipAddress, port, DefaultServerId)
        {
        }

        /// <summary>
        /// Initializes a new instance of the RpcChannel class.
        /// </summary>
        /// <param name="transport">An RpcSocketTransport object.</param>
        internal RpcChannel(RpcSocketTransport transport)
        {
            if (transport == null)
            {
                throw new ArgumentNullException("transport");
            }

            this.Initialize(transport);
        }

        /// <summary>
        /// This event is raised any time that an expected behavior is encountered by the RpcChannel object.
        /// Examples include, receiving malformed messages from the server.  This event should be used primarily for
        /// debug purposes and should only be raised in rare occasions.
        /// </summary>
        public event EventHandler<RpcErrorEventArgs> UnexpectedErrorOccurred;

        /// <summary>
        /// Gets or sets the amount of time to wait for remote operations to complete. OperationTimeout can only be set to System.Threading.Timeout.InfiniteTimeSpan or a TimeSpan value with total milliseconds greater or equal to 0 and less than 2147483648.
        /// </summary>
        public TimeSpan OperationTimeout
        {
            get
            {
                return this.transport.OperationTimeout;
            }

            set
            {
                this.transport.OperationTimeout = value;
            }
        }

        /// <summary>
        /// Gets a value indicating whether or not this RpcChannel is connected to an RpcServer.
        /// </summary>
        public bool IsConnected
        {
            get
            {
                bool connected = false;

                try
                {
                    this.connectLock.EnterReadLock();

                    connected = this.transport.IsConnected;
                }
                finally
                {
                    this.connectLock.ExitReadLock();
                }

                return connected;
            }
        }

        /// <summary>
        /// Gets the platform of the currently connected server, or null if not connected.
        /// </summary>
        internal string Platform { get; private set; }

        /// <summary>
        /// Gets the architecture of the currently connected server, or null if not connected.
        /// </summary>
        internal string Architecture { get; private set; }

        /// <summary>
        /// Invokes a remote method exposed by an RpcServer.
        /// </summary>
        /// <param name="methodName">The name of the method to invoke.</param>
        /// <param name="methodParameters">The parameters to be passed to the remote method.</param>
        /// <param name="invokeTimeout">Overrides the channel timeout for this remote method call.</param>
        /// <returns>An RpcArchive that contains all of the data returned by the remote method.</returns>
        /// <exception cref="RpcException">There was a problem with RPC protocol or remote method invocation.</exception>
        /// <exception cref="TimeoutException">The server did not respond before the specified timeout elapsed.</exception>
        /// <remarks>
        /// The TimeoutException is the most common exception to be thrown by this method. Possible reasons include:
        /// - The hook being called is blocking for longer than the time-out specified for the method call
        /// - A hook call made earlier is still executing on the server and blocking it from processing other requests
        /// - The server might not be processing requests (RpcServer::ProcessPendingRequests is not getting called)
        /// - The time between calls to ProcessPendingRequests on the server exceeds the time out for the method call
        /// - The server is unable to handle the current volume of traffic because the blocking time of each request
        ///   adds up to cause some of the requests to time out.
        /// </remarks>
        public RpcArchive InvokeRemoteMethod(string methodName, RpcArchive methodParameters, TimeSpan invokeTimeout)
        {
            try
            {
                this.connectLock.EnterReadLock();

                return this.InvokeRemoteMethodImpl(methodName, methodParameters, invokeTimeout);
            }
            finally
            {
                this.connectLock.ExitReadLock();
            }
        }

        /// <summary>
        /// Invokes a remote method exposed by an RpcServer.
        /// </summary>
        /// <param name="methodName">The name of the method to invoke.</param>
        /// <param name="methodParameters">The parameters to be passed to the remote method.</param>
        /// <returns>An RpcArchive that contains all of the data returned by the remote method.</returns>
        /// <exception cref="RpcException">There was a problem with RPC protocol or remote method invocation.</exception>
        /// <exception cref="TimeoutException">The server did not respond before the channel timeout elapsed.</exception>
        /// <remarks>
        /// The TimeoutException is the most common exception to be thrown by this method. Possible reasons include:
        /// - The hook being called is blocking for longer than the time-out specified for the channel
        /// - A hook call made earlier is still executing on the server and blocking it from processing other requests
        /// - The server might not be processing requests (RpcServer::ProcessPendingRequests is not getting called)
        /// - The time between calls to ProcessPendingRequests on the server exceeds the time out for the method call
        /// - The server is unable to handle the current volume of traffic because the blocking time of each request
        ///   adds up to cause some of the requests to time out.
        /// </remarks>
        public RpcArchive InvokeRemoteMethod(string methodName, RpcArchive methodParameters)
        {
            try
            {
                this.connectLock.EnterReadLock();

                return this.InvokeRemoteMethodImpl(methodName, methodParameters, null);
            }
            finally
            {
                this.connectLock.ExitReadLock();
            }
        }

        /// <summary>
        /// Gets a list of registered hooks.
        /// </summary>
        /// <returns>The list of hooks registered on the server.</returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1024:UsePropertiesWhereAppropriate", Justification = "Underlying information gets refreshed every call.")]
        public IEnumerable<RpcRemoteMethod> GetRemoteMethods()
        {
            try
            {
                RpcArchive remoteMethods = this.InvokeRemoteMethod(BuiltInHookNames.EnumRegisteredHooks, new RpcArchive());

                string[] builtInHooks =
                {
                    BuiltInHookNames.UpdateEventSubscription,
                    BuiltInHookNames.VerifyPropertiesOnConnect,
                    BuiltInHookNames.SetMaximumQueueSize,
                    BuiltInHookNames.EnumRegisteredHooks
                };

                IEnumerable<RpcRemoteMethod> allMethods =
                    from h in remoteMethods.GetNamedCollection<RpcArchive>("hooks")
                    select new RpcRemoteMethod(h.GetNamedValue<string>("name"));

                return from h in allMethods where !builtInHooks.Contains(h.Name) select h;
            }
            catch (RpcRemoteMethodNotFoundException)
            {
                return new RpcRemoteMethod[0];
            }
        }

        /// <summary>
        /// Connect to an RpcServer.
        /// </summary>
        /// <exception cref="RpcException">Thrown if the RpcChannel is not able to connect to an RpcServer.</exception>
        public void Connect()
        {
            this.ConnectImpl(null);
        }

        /// <summary>
        /// Connect to an RpcServer.
        /// </summary>
        /// <param name="timeout">Overrides the channel timeout for the connection.</param>
        /// <exception cref="RpcException">Thrown if the RpcChannel is not able to connect to an RpcServer.</exception>
        public void Connect(TimeSpan timeout)
        {
            this.ConnectImpl(timeout);
        }

        /// <summary>
        /// Disconnect from an RpcServer.
        /// </summary>
        public void Disconnect()
        {
            this.ThrowIfDisposed();
            try
            {
                if (!this.connectLock.IsWriteLockHeld)
                {
                    this.connectLock.EnterWriteLock();
                }

                if (this.connectionId != Guid.Empty)
                {
                    RpcClientEventSource.Logger.Disconnect(this.connectionId);

                    this.connectionId = Guid.Empty;
                }

                // Cancelling the CancellationTokenSource will signal the read operation that it needs to stop
                // listening for incoming messages.
                if (this.cancelReadingTokenSource != null)
                {
                    this.cancelReadingTokenSource.Cancel();
                }

                // Now wait until the reading task has completed and report any exceptions that it captured.
                if (this.readingIncomingMessagesTask != null)
                {
                    try
                    {
                        this.readingIncomingMessagesTask.Wait();
                    }
                    catch (AggregateException ex)
                    {
                        // If we get anything other than a notification that the task has been canceled then throw.
                        if (!(ex.InnerException is TaskCanceledException))
                        {
                            throw ex.InnerException;
                        }
                    }

                    this.readingIncomingMessagesTask = null;
                }

                // We're all done using the CancellationTokenSource so dispose of it.
                if (this.cancelReadingTokenSource != null)
                {
                    this.cancelReadingTokenSource.Dispose();
                    this.cancelReadingTokenSource = null;
                }

                this.transport.Disconnect();

                this.Platform = null;
                this.Architecture = null;
            }
            finally
            {
                if (this.connectLock.IsWriteLockHeld)
                {
                    this.connectLock.ExitWriteLock();
                }
            }
        }

        /// <summary>
        /// Subscribes to an RPC event. 
        /// This method adds a new entry to the RPC server event subscriptions table - a unique combination of the event name and a delegate.
        /// </summary>
        /// <param name="eventName">The RPC event name.</param>
        /// <param name="eventHandler">The delegate that is called when an RPC event happens and event data is received on the client side.</param>
        /// <returns>The Id of added subscription entry.</returns>
        public Guid Subscribe(string eventName, EventHandler<RpcEventArgs> eventHandler)
        {
            if (string.IsNullOrEmpty(eventName))
            {
                throw new ArgumentNullException("eventName");
            }

            if (eventHandler == null)
            {
                throw new ArgumentNullException("eventHandler");
            }

            RpcClientEventSource.Logger.Subscribe();

            ConcurrentDictionary<Guid, EventHandler<RpcEventArgs>> subscribedEventHandlers;

            if (!this.serverEventSubscriptions.TryGetValue(eventName, out subscribedEventHandlers))
            {
                subscribedEventHandlers = new ConcurrentDictionary<Guid, EventHandler<RpcEventArgs>>();
                this.serverEventSubscriptions.AddOrUpdate(eventName, subscribedEventHandlers, (key, oldvalue) => oldvalue = subscribedEventHandlers);
            }

            Guid guid = Guid.NewGuid();
            subscribedEventHandlers.AddOrUpdate(guid, eventHandler, (key, oldvalue) => oldvalue = eventHandler);

            if (this.transport.IsConnected)
            {
                try
                {
                    this.UpdateEventSubscriptionOnServer(eventName, true);
                }
                catch (Exception)
                {
                    // If the subscribe operation fails, the exception will be raised to the user.
                    // Remove the handler from the dictionary.
                    EventHandler<RpcEventArgs> removed;
                    subscribedEventHandlers.TryRemove(guid, out removed);

                    if (!subscribedEventHandlers.Any())
                    {
                        ConcurrentDictionary<Guid, EventHandler<RpcEventArgs>> removedSubItem;
                        this.serverEventSubscriptions.TryRemove(eventName, out removedSubItem);
                    }

                    throw;
                }
            }

            return guid;
        }

        /// <summary>
        /// Unsubscribes from an RPC event.
        /// This method removes a corresponding entry from the RPC server event subscriptions table.
        /// </summary>
        /// <param name="subscriptionEntryId">The Id of a subscription entry obtained from Subscribe method.</param>
        public void Unsubscribe(Guid subscriptionEntryId)
        {
            var nullEventEntry = default(KeyValuePair<string, ConcurrentDictionary<Guid, EventHandler<RpcEventArgs>>>);
            var eventEntry = this.serverEventSubscriptions.SingleOrDefault(e => e.Value.ContainsKey(subscriptionEntryId));
            if (eventEntry.Equals(nullEventEntry))
            {
                throw new RpcNotSubscribedException("The RPC server event subscriptions table doesn't contain an entry with specified Id");
            }

            ConcurrentDictionary<Guid, EventHandler<RpcEventArgs>> subscribedEventHandlers = eventEntry.Value;
            EventHandler<RpcEventArgs> removed;
            subscribedEventHandlers.TryRemove(subscriptionEntryId, out removed);

            if (!subscribedEventHandlers.Any())
            {
                if (this.IsConnected)
                {
                    this.UpdateEventSubscriptionOnServer(eventEntry.Key, false);
                }

                ConcurrentDictionary<Guid, EventHandler<RpcEventArgs>> removedTopLevelItem;
                this.serverEventSubscriptions.TryRemove(eventEntry.Key, out removedTopLevelItem);
            }
        }

        /// <summary>
        /// Register the <see cref="IExceptionFactory"/> that will be used to map user-defined
        /// error codes to exceptions.
        /// </summary>
        /// <param name="factory">Factory that will be used to do code-to-exception mapping.</param>
        public void RegisterExceptionFactory(IExceptionFactory factory)
        {
            RpcClientEventSource.Logger.RegisterExceptionFactory();
            this.exceptionFactory = factory;
        }

        /// <summary>
        /// Disposes of the managed resources used by this class.
        /// </summary>
        protected override void DisposeManagedResources()
        {
            this.Disconnect();
        }

        private static T DeserializeObject<T>(DataContractJsonSerializer serializer, string response)
        {
            var responseBuffer = Encoding.UTF8.GetBytes(response);

            using (MemoryStream responseStream = new MemoryStream(responseBuffer))
            {
                return (T)serializer.ReadObject(responseStream);
            }
        }

        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2202:Do not dispose objects multiple times", Justification = "The StreamReader will not dispose of the stream.")]
        private static string SerializeObject<T>(DataContractJsonSerializer serializer, T message)
        {
            using (MemoryStream stream = new MemoryStream())
            {
                serializer.WriteObject(stream, message);

                stream.Position = 0;
                using (StreamReader reader = new StreamReader(stream))
                {
                    return reader.ReadToEnd();
                }
            }
        }

        /// <summary>
        /// Implementation of the connecting to an RpcServer.
        /// </summary>
        /// <param name="timeout">Overrides the channel timeout for the connection.</param>
        /// <exception cref="RpcException">Thrown if the RpcChannel is not able to connect to an RpcServer.</exception>
        private void ConnectImpl(TimeSpan? timeout)
        {
            this.ThrowIfDisposed();

            RpcSocketTransport.ThrowIfInvalidTimeout(timeout, "timeout");
            try
            {
                DateTime startTime = DateTime.Now;
                this.connectLock.EnterWriteLock();

                this.connectionId = Guid.NewGuid();
                RpcClientEventSource.Logger.TryConnect(
                    this.connectionId,
                    this.transport.ServerId == DefaultServerId,
                    this.transport.Port == RpcSocketTransport.DefaultPort,
                    this.transport.OperationTimeout.Ticks);

                if (this.transport.IsConnected)
                {
                    return;
                }

                TimeSpan? timeRemaining = this.CalculateRemainingTime(timeout, startTime);

                this.transport.Connect(timeRemaining);

                // Create a new CancellationTokenSource so that each time that we call this.transport.GetIncomingSerializedMessages
                // we will pass it a new CancellationToken.
                this.cancelReadingTokenSource = new CancellationTokenSource();

                this.readingIncomingMessagesTask = Task.Run(new Action(this.ProcessIncomingMessages));

                this.readingIncomingMessagesTask.ContinueWith(
                    t =>
                    {
                        this.RaiseUnexpectedErrorOccurredEvent(new RpcErrorEventArgs("Unexpected error ocurred on RpcChannel listening thread.  The channel will be disconnected.", null, t.Exception));
                        this.Disconnect();
                    },
                    TaskContinuationOptions.OnlyOnFaulted);

                try
                {
                    RpcArchive inputParameters = new RpcArchive();
                    inputParameters.SetNamedCollection<string>("events", this.serverEventSubscriptions.Keys);

                    timeRemaining = this.CalculateRemainingTime(timeout, startTime);

                    RpcArchive serverInfo = this.InvokeRemoteMethodImpl(BuiltInHookNames.VerifyPropertiesOnConnect, inputParameters, timeRemaining);

                    if (serverInfo != null)
                    {
                        if (serverInfo.HasNamedValue(PlatformValue))
                        {
                            this.Platform = serverInfo.GetNamedValue<string>(PlatformValue);
                        }

                        if (serverInfo.HasNamedValue(ArchitectureValue))
                        {
                            this.Architecture = serverInfo.GetNamedValue<string>(ArchitectureValue);
                        }
                    }

                    RpcClientEventSource.Logger.Connected(
                        this.connectionId,
                        this.Platform != null ? this.Platform : string.Empty,
                        this.Architecture != null ? this.Architecture : string.Empty);
                }
                catch (RpcServerIdNotFoundException)
                {
                    // Pass through actual connection errors as-is
                    this.Disconnect();
                    throw;
                }
                catch (Exception ex)
                {
                    // If the call to InvokeRemoteMethod fails, we are not successfully connected.
                    this.Disconnect();

                    RpcException connectException = new RpcException("Failed to retrieve server info.", ex);
                    this.ThrowExceptionsOnPendingRequests(connectException);
                    throw connectException;
                }
            }
            finally
            {
                if (this.connectLock.IsWriteLockHeld)
                {
                    this.connectLock.ExitWriteLock();
                }
            }
        }

        private void ProcessIncomingMessages()
        {
            string lastMessage = null;

            try
            {
                foreach (string incomingSerializedMessage in this.transport.GetIncomingSerializedMessages(this.cancelReadingTokenSource.Token))
                {
                    lastMessage = incomingSerializedMessage;

                    this.ProcessIncomingSerializedMessage(incomingSerializedMessage);
                }
            }
            catch (RpcException ex)
            {
                this.ThrowExceptionsOnPendingRequests(ex);
                this.RaiseUnexpectedErrorOccurredEvent(new RpcErrorEventArgs("Failure in thread that processes incoming messages. Message processing has stopped.", lastMessage, ex));
            }
        }

        private TimeSpan? CalculateRemainingTime(TimeSpan? timeout, DateTime startTime)
        {
            if (timeout == null)
            {
                return null;
            }

            TimeSpan operationTimeout = timeout.Value;
            TimeSpan timeRemaining;

            if (operationTimeout == Timeout.InfiniteTimeSpan)
            {
                timeRemaining = operationTimeout;
            }
            else
            {
                TimeSpan timePassed = DateTime.Now - startTime;
                timeRemaining = operationTimeout - timePassed;
                if (timeRemaining <= TimeSpan.Zero)
                {
                    timeRemaining = TimeSpan.FromMilliseconds(1);
                }
            }

            return timeRemaining;
        }

        private void Initialize(RpcSocketTransport internalTransport)
        {
            this.serverEventSubscriptions = new ConcurrentDictionary<string, ConcurrentDictionary<Guid, EventHandler<RpcEventArgs>>>();
            this.transport = internalTransport;

            this.responseDeserializer = new DataContractJsonSerializer(typeof(RpcResponse), JsonSerializerKnownTypes);
            this.messageSerializer = new DataContractJsonSerializer(typeof(RpcMessage), JsonSerializerKnownTypes);

            this.OperationTimeout = DefaultOperationTimeout;

            this.pendingRequests = new ConcurrentDictionary<long, AsyncResult<RpcResponse>>();
        }

        private string SerializeRequest(RpcRequest request)
        {
            DataContractJsonSerializerSettings settings = new DataContractJsonSerializerSettings();
            settings.EmitTypeInformation = EmitTypeInformation.Never;
            settings.KnownTypes = JsonSerializerKnownTypes.Concat(request.Parameters.GetAllTypes());
            DataContractJsonSerializer requestSerializer = new DataContractJsonSerializer(typeof(RpcRequest), settings);

            string serializedRequest = SerializeObject<RpcRequest>(requestSerializer, request);
            RpcMessage requestMessage = new RpcMessage() { MessageType = RpcMessageType.Request, Payload = serializedRequest };
            return SerializeObject<RpcMessage>(this.messageSerializer, requestMessage);
        }

        /// <summary>
        /// Invokes a remote method exposed by an RpcServer.
        /// </summary>
        /// <param name="methodName">The name of the method to invoke.</param>
        /// <param name="methodParameters">The parameters to be passed to the remote method.</param>
        /// <param name="invokeTimeout">Overrides the channel timeout for this remote method call if not null.</param>
        /// <returns>An RpcArchive that contains all of the data returned by the remote method.</returns>
        private RpcArchive InvokeRemoteMethodImpl(string methodName, RpcArchive methodParameters, TimeSpan? invokeTimeout)
        {
            this.ThrowIfDisposed();

            if (methodParameters == null)
            {
                throw new ArgumentNullException("methodParameters");
            }

            if (string.IsNullOrWhiteSpace(methodName))
            {
                throw new ArgumentNullException("methodName");
            }

            RpcSocketTransport.ThrowIfInvalidTimeout(invokeTimeout, "invokeTimeout");

            // Don't check IsConnected until Connect() returns on all threads
            if (!this.transport.IsConnected)
            {
                throw new RpcChannelDisconnectedException("Cannot invoke a remote method if the RpcChannel is not connected.");
            }

            DateTime startTime = DateTime.Now;
            RpcRequest request = new RpcRequest() { MethodName = methodName, Parameters = methodParameters, RequestId = this.GetNextRequestId() };

            string serializedRequestMessage = null;
            try
            {
                serializedRequestMessage = this.SerializeRequest(request);
            }
            catch (SerializationException ex)
            {
                throw new RpcException("Failed to serialize parameters to remote method.", ex);
            }

            RpcClientEventSource.Logger.InvokeRemoteMethod(
                this.connectionId,
                methodParameters.ArchiveGuid,
                methodName.Length,
                methodParameters.ParameterCount,
                serializedRequestMessage.Length * sizeof(char));

            AsyncResult<RpcResponse> asyncResult = new AsyncResult<RpcResponse>();

            try
            {
                this.pendingRequests.AddOrUpdate(request.RequestId, asyncResult, (key, oldValue) => oldValue = asyncResult);

                if (invokeTimeout.HasValue)
                {
                    this.transport.SendRequest(serializedRequestMessage, invokeTimeout.Value);
                }
                else
                {
                    this.transport.SendRequest(serializedRequestMessage);
                }

                TimeSpan operationTimeout = invokeTimeout == null ? this.OperationTimeout : invokeTimeout.Value;
                TimeSpan timeRemaining;

                if (operationTimeout == Timeout.InfiniteTimeSpan)
                {
                    timeRemaining = operationTimeout;
                }
                else
                {
                    TimeSpan timePassed = DateTime.Now - startTime;
                    timeRemaining = operationTimeout - timePassed;
                    if (timeRemaining < TimeSpan.Zero)
                    {
                        timeRemaining = TimeSpan.Zero;
                    }
                }

                RpcResponse response = asyncResult.EndInvoke(timeRemaining);

                RpcExceptionFactory.ThrowExceptionForResponseCodes(request, response, this.exceptionFactory);

                return response.Archive;
            }
            catch (TimeoutException)
            {
                // Throw a custom time-out exception in place of the default one to offer more troubleshooting information.
                throw new TimeoutException(string.Format(CultureInfo.InvariantCulture, "Remote method call timed out. '{0}' or a previous call might be slow to return, or the server is not processing requests often enough.", methodName));
            }
            finally
            {
                AsyncResult<RpcResponse> removed;
                this.pendingRequests.TryRemove(request.RequestId, out removed);
            }
        }

        /// <summary>
        /// Gets the next sequential request Id.
        /// </summary>
        /// <returns>The next sequential request Id.</returns>
        private uint GetNextRequestId()
        {
            return (uint)Interlocked.Increment(ref lastRequestId);
        }

        private void UpdateEventSubscriptionOnServer(string eventName, bool isSubscribed)
        {
            RpcArchive archive = new RpcArchive();
            archive.SetNamedValue("eventName", eventName);
            archive.SetNamedValue("subscribe", isSubscribed);
            this.InvokeRemoteMethod(BuiltInHookNames.UpdateEventSubscription, archive);
        }

        /// <summary>
        /// This relatively simple function exists so that it can be mocked in the unit tests.
        /// </summary>
        /// <param name="serializedRpcResponse">A serialized RpcResponse message.</param>
        /// <returns>A deserialized RpcResponse object.</returns>
        private RpcResponse DeserializeResponse(string serializedRpcResponse)
        {
            return DeserializeObject<RpcResponse>(this.responseDeserializer, serializedRpcResponse);
        }

        private void RaiseUnexpectedErrorOccurredEvent(RpcErrorEventArgs args)
        {
            var temp = this.UnexpectedErrorOccurred;
            if (temp != null)
            {
                temp(this, args);
            }
        }

        private void ProcessIncomingSerializedMessage(string incomingSerializedMessage)
        {
            RpcMessage rpcMessage = null;
            try
            {
                rpcMessage = DeserializeObject<RpcMessage>(this.messageSerializer, incomingSerializedMessage);
            }
            catch (SerializationException ex)
            {
                this.RaiseUnexpectedErrorOccurredEvent(new RpcErrorEventArgs("Failed to deserialize incoming RpcMessage. The message will be dropped. If this was a response to a request then that request will timeout.", incomingSerializedMessage, ex));
            }

            if (rpcMessage != null)
            {
                switch (rpcMessage.MessageType)
                {
                    case RpcMessageType.Response:
                        if (this.pendingRequests.Count > 0)
                        {
                            try
                            {
                                RpcResponse response = this.DeserializeResponse(rpcMessage.Payload);

                                AsyncResult<RpcResponse> value;
                                if (this.pendingRequests.TryGetValue(response.RequestId, out value))
                                {
                                    value.SetAsCompleted(response, false);
                                }
                            }
                            catch (SerializationException ex)
                            {
                                // Include incoming serialized message in all exceptions thrown to make future diagnostics easier.
                                this.RaiseUnexpectedErrorOccurredEvent(new RpcErrorEventArgs("Failed to deserialize incoming response. See inner exception for details.", incomingSerializedMessage, ex));
                                this.ThrowExceptionsOnPendingRequests(new RpcException(string.Format(CultureInfo.CurrentCulture, "Failed to deserialize incoming response:{0}{1}{2}See inner exception for details.", Environment.NewLine, incomingSerializedMessage, Environment.NewLine), ex));
                            }
                        }
                        else
                        {
                            this.RaiseUnexpectedErrorOccurredEvent(new RpcErrorEventArgs("An unexpected response was received despite no pending request. The message will be ignored.", rpcMessage.Payload, null));
                        }

                        break;
                    case RpcMessageType.Request:
                        try
                        {
                            RpcRequest request = DeserializeObject<RpcRequest>(new DataContractJsonSerializer(typeof(RpcRequest), JsonSerializerKnownTypes), rpcMessage.Payload);

                            ConcurrentDictionary<Guid, EventHandler<RpcEventArgs>> eventSubscriptions;

                            if (this.serverEventSubscriptions.TryGetValue(request.MethodName, out eventSubscriptions))
                            {
                                if (!eventSubscriptions.IsEmpty)
                                {
                                    // Running subscriptions on a task so that we don't block waiting
                                    // for the event handlers before we can process further incoming 
                                    // messages.
                                    Task.Run(() =>
                                    {
                                        RpcClientEventSource.Logger.SubscriptionMethodInvoked(
                                            this.connectionId,
                                            request.Parameters.ArchiveGuid,
                                            request.MethodName.Length,
                                            eventSubscriptions.Count);
                                        
                                        foreach (var subscription in eventSubscriptions.Values)
                                        {
                                            try
                                            {
                                                subscription(this, new RpcEventArgs(request.MethodName, request.Parameters));
                                            }
                                            catch (Exception e)
                                            {
                                                this.RaiseUnexpectedErrorOccurredEvent(new RpcErrorEventArgs(string.Format(CultureInfo.InvariantCulture, "Event handler threw an unexpected exception. Event name: {0}", request.MethodName), null, e));
                                            }
                                        }
                                    });
                                }
                            }
                            else
                            {
                                this.RaiseUnexpectedErrorOccurredEvent(new RpcErrorEventArgs(string.Format(CultureInfo.InvariantCulture, "Received an event notification for an event that the client is not subscribed to. Event name: {0}", request.MethodName), null, null));
                            }
                        }
                        catch (SerializationException ex)
                        {
                            this.RaiseUnexpectedErrorOccurredEvent(new RpcErrorEventArgs("Failed to deserialize incoming serialized message.", incomingSerializedMessage, ex));
                        }

                        break;
                }
            }
        }

        private void ThrowExceptionsOnPendingRequests(RpcException ex)
        {
            foreach (var request in this.pendingRequests.Values)
            {
                request.SetAsCompleted(ex, false);
            }

            this.pendingRequests.Clear();
        }
    }
}
